package com.xiaojie.hadoop.mapper;

import com.xiaojie.hadoop.bean.CommodityBean;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author 熟透的蜗牛
 * @version 1.0
 * @description: 对商品进行mapper
 * @date 2025/1/7 20:03
 */
public class CommodityMapper extends TableMapper<Text, CommodityBean> {

//            | 9905689326371 | 45896542     | 100         | 女士带花连衣裙  | 99.9  | test       | 1131001  | 16:51:05   |
//            | 9906689326371 | 45896542     | 100         | 女士带花连衣裙  | 99.9  | test       | 2281001  | 16:51:06   |
//            | 9909689326371 | 1563521      | 25696       | 女士带花连衣裙2 | 59.9  | test111    | 6810001  | 16:51:09   |
//            | 9925689326371 | 45896542     | 100         | 女士带花连衣裙  | 99.9  | test       | 1331001  | 16:51:05   |
//            | 9947689326371 | 45896542     | 100         | 女士带花连衣裙  | 99.9  | test       | 3711001  | 16:51:07   |
//            | 9957689326371 | 45896542     | 100         | 女士带花连衣裙  | 99.9  | test       | 3841001  | 16:51:07   |


    @Override
    protected void map(ImmutableBytesWritable key, Result value, Mapper<ImmutableBytesWritable, Result, Text, CommodityBean>.Context context) throws IOException, InterruptedException {
        Text textOutKey = new Text();
        CommodityBean commodityBean = new CommodityBean();
        for (Cell cell : value.rawCells()) {
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            if ("commodity_id".equals(qualifier)) {
                String commodityId = Bytes.toString(CellUtil.cloneValue(cell));
                commodityBean.setCommodityId(commodityId);
                //设置商品id作为输出键
                textOutKey.set(Bytes.toString(CellUtil.cloneValue(cell)));
            } else if ("name".equals(qualifier)) {
                String commodityName = Bytes.toString(CellUtil.cloneValue(cell));
                commodityBean.setCommodityName(commodityName);
            } else if ("price".equals(qualifier)) {
                String price = Bytes.toString(CellUtil.cloneValue(cell));
                commodityBean.setPrice(price);
            } else if ("user_id".equals(qualifier)) {
                commodityBean.setUserId(Bytes.toString(CellUtil.cloneValue(cell)));
            }else if ("picture".equals(qualifier)) {
                String picture = Bytes.toString(CellUtil.cloneValue(cell));
                commodityBean.setPicture(picture);
            }
            if (!checkCommodityBean(commodityBean)) {
                continue;
            }
            commodityBean.setCount(1l);
            context.write(textOutKey, commodityBean);
        }
    }

    /**
     * @param commodityBean
     * @description: 校验数据完整性
     * @return: boolean
     * @author 熟透的蜗牛
     * @date: 2025/1/8 5:18
     */
    private boolean checkCommodityBean(CommodityBean commodityBean) {
        if (commodityBean.getCommodityId() != null && commodityBean.getCommodityName() != null
                && commodityBean.getPrice() != null &&commodityBean.getPicture()!=null && commodityBean.getUserId() != null) {
            return true;
        }
        return false;
    }
}